/**
 * Copyright 1999-2015 dangdang.com.
 * <p>
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 * </p>
 */

package spring.cloud.tasks.api.service.impl;

import com.google.common.base.Strings;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import jxl.Cell;
import jxl.CellType;
import jxl.Sheet;
import jxl.Workbook;
import jxl.write.*;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.zookeeper.data.Stat;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;
import org.springframework.web.multipart.MultipartFile;
import spring.cloud.tasks.api.database.entity.TaskConfigurationDto;
import spring.cloud.tasks.api.database.service.TaskConfigurationDtoService;
import spring.cloud.tasks.api.domain.*;
import spring.cloud.tasks.api.service.TaskService;
import spring.cloud.tasks.api.service.ZooKeeperService;
import spring.cloud.tasks.api.utils.*;
import spring.cloud.tasks.api.vo.GetTaskConfigurationVo;
import spring.cloud.tasks.api.zookeeper.TransactionalZooKeeper;
import spring.cloud.tasks.api.zookeeper.ZooKeeper;

import javax.annotation.Resource;
import java.io.File;
import java.lang.Boolean;
import java.text.ParseException;
import java.util.*;

@Slf4j
public class TaskServiceImpl implements TaskService {

    public static final String CONFIG_ITEM_LOAD_LEVEL = "loadLevel";
    public static final String CONFIG_ITEM_ENABLED = "enabled";
    public static final String CONFIG_ITEM_DESCRIPTION = "description";
    public static final String CONFIG_ITEM_CUSTOM_CONTEXT = "customContext";
    public static final String CONFIG_ITEM_JOB_TYPE = "jobType";

    public static final String CONFIG_ITEM_SHARDING_ITEM_PARAMETERS = "shardingItemParameters";
    public static final String CONFIG_ITEM_JOB_PARAMETER = "jobParameter";

    public static final String CONFIG_ITEM_FAILOVER = "failover";
    public static final String CONFIG_ITEM_MONITOR_EXECUTION = "monitorExecution";
    public static final String CONFIG_ITEM_TIMEOUT_4_ALARM_SECONDS = "timeout4AlarmSeconds";
    public static final String CONFIG_ITEM_TIMEOUT_SECONDS = "timeoutSeconds";
    public static final String CONFIG_DISABLE_TIMEOUT_SECONDS = "disableTimeoutSeconds";
    public static final String CONFIG_ITEM_TIME_ZONE = "timeZone";
    public static final String CONFIG_ITEM_CRON = "cron";
    public static final String CONFIG_ITEM_PAUSE_PERIOD_DATE = "pausePeriodDate";
    public static final String CONFIG_ITEM_PAUSE_PERIOD_TIME = "pausePeriodTime";
    public static final String CONFIG_ITEM_PROCESS_COUNT_INTERVAL_SECONDS = "processCountIntervalSeconds";
    public static final String CONFIG_ITEM_SHARDING_TOTAL_COUNT = "shardingTotalCount";
    public static final String CONFIG_ITEM_SHOW_NORMAL_LOG = "showNormalLog";
    public static final String CONFIG_ITEM_JOB_DEGREE = "jobDegree";
    public static final String CONFIG_ITEM_ENABLED_REPORT = "enabledReport";
    public static final String CONFIG_ITEM_PREFER_LIST = "preferList";
    public static final String CONFIG_ITEM_USE_DISPREFER_LIST = "useDispreferList";


    public static final String CONFIG_ITEM_JOB_CLASS = "jobClass";
    public static final String CONFIG_ITEM_RERUN = "rerun";


    private static final String ERR_MSG_PENDING_STATUS = "job:[{}] item:[{}] on executor:[{}] execution status is "
            + "PENDING as {}";
    private static final String ERR_MSG_TOO_LONG_TO_DISPLAY = "Not display the log as the length is out of max length";
    private final Random random = new Random();

    @Resource
    private ZooKeeperService zooKeeperService;
    @Resource
    private TaskConfigurationDtoService taskConfigurationDtoService;


    private TaskStatus getTaskStatus(final String taskId, ZooKeeper zooKeeper, boolean enabled) {
        // see if all the shards is finished.
        boolean isAllShardsFinished = isAllShardsFinished(taskId, zooKeeper);
        if (enabled) {
            if (isAllShardsFinished) {
                return TaskStatus.READY;
            }
            return TaskStatus.RUNNING;
        } else {
            if (isAllShardsFinished) {
                return TaskStatus.STOPPED;
            }
            return TaskStatus.STOPPING;
        }
    }

    private boolean isAllShardsFinished(final String taskId, ZooKeeper zooKeeper) {
        List<String> executionItems = zooKeeper.getChildren(TaskNodePath.getExecutionNodePath(taskId));
        boolean isAllShardsFinished = true;
        if (executionItems != null && !executionItems.isEmpty()) {
            for (String itemStr : executionItems) {
                boolean isItemCompleted = zooKeeper.checkExists(TaskNodePath.getExecutionNodePath(taskId, itemStr, "completed"));
                boolean isItemRunning = zooKeeper.checkExists(TaskNodePath.getExecutionNodePath(taskId, itemStr, "running"));
                if (!isItemCompleted && isItemRunning) {
                    isAllShardsFinished = false;
                    break;
                }
            }
        }
        return isAllShardsFinished;
    }

    @Transactional(rollbackFor = Exception.class)
    @Override
    public void enableJob(String nameSpace, String taskId, String updatedBy) throws RuntimeException {
        TaskConfigurationDto taskConfigurationDto = taskConfigurationDtoService.getTaskConfigurationDtoByNameSpaceAndTaskId(nameSpace, taskId);
        if (taskConfigurationDto == null) {
            throw new RuntimeException("不能启用该作业（" + taskId + "），因为该作业不存在");
        }
        if (taskConfigurationDto.getEnabled()) {
            throw new RuntimeException("该作业（" + taskId + "）已经处于启用状态");
        }
        ZooKeeper zooKeeper = zooKeeperService.getZooKeeper(nameSpace);
        boolean allShardsFinished = isAllShardsFinished(taskId, zooKeeper);
        if (!allShardsFinished) {
            throw new RuntimeException("不能启用该作业（" + taskId + "），因为该作业不处于STOPPED状态");
        }
        taskConfigurationDto.setEnabled(true);
        taskConfigurationDto.setLastUpdateTime(new Date());
        taskConfigurationDto.setLastUpdateBy(updatedBy);
        taskConfigurationDtoService.updateByPrimaryKey(taskConfigurationDto);
        zooKeeper.update(TaskNodePath.getConfigNodePath(taskId, CONFIG_ITEM_ENABLED), true);
    }

    @Transactional(rollbackFor = Exception.class)
    @Override
    public void disableJob(String nameSpace, String taskId, String updatedBy) throws RuntimeException {
        TaskConfigurationDto taskConfigurationDto = taskConfigurationDtoService.getTaskConfigurationDtoByNameSpaceAndTaskId(nameSpace, taskId);
        if (taskConfigurationDto == null) {
            throw new RuntimeException("不能禁用该作业（" + taskId + "），因为该作业不存在");
        }
        if (!taskConfigurationDto.getEnabled()) {
            throw new RuntimeException("该作业（" + taskId + "）已经处于禁用状态");
        }
        taskConfigurationDto.setEnabled(Boolean.FALSE);
        taskConfigurationDto.setLastUpdateTime(new Date());
        taskConfigurationDto.setLastUpdateBy(updatedBy);
        taskConfigurationDtoService.updateByPrimaryKey(taskConfigurationDto);
        ZooKeeper zooKeeper = zooKeeperService.getZooKeeper(nameSpace);
        zooKeeper.update(TaskNodePath.getConfigNodePath(taskId, CONFIG_ITEM_ENABLED), false);
    }

    @Transactional(rollbackFor = Exception.class)
    @Override
    public void removeTask(String nameSpace, String taskId) throws RuntimeException {
        TaskConfigurationDto taskConfigurationDto = taskConfigurationDtoService.getTaskConfigurationDtoByNameSpaceAndTaskId(nameSpace, taskId);
        if (taskConfigurationDto == null) {
            throw new RuntimeException("不能删除该作业（" + taskId + "），因为该作业不存在");
        }

        ZooKeeper zooKeeper = zooKeeperService.getZooKeeper(nameSpace);
        TaskStatus taskStatus = getTaskStatus(taskId, zooKeeper, taskConfigurationDto.getEnabled());

        if (TaskStatus.STOPPED != taskStatus) {
            throw new RuntimeException(
                    String.format("不能删除该作业(%s)，因为该作业不处于STOPPED状态", taskId));
        }

        Stat stat = zooKeeper.getStat(TaskNodePath.getTaskNodePath(taskId));
        if (stat != null) {
            long createTimeDiff = System.currentTimeMillis() - stat.getCtime();
            if (createTimeDiff < SaturnConstants.JOB_CAN_BE_DELETE_TIME_LIMIT) {
                throw new RuntimeException(String.format("不能删除该作业(%s)，因为该作业创建时间距离现在不超过%d分钟", taskId, SaturnConstants.JOB_CAN_BE_DELETE_TIME_LIMIT / 60000));
            }
        }
        // remove job from db
        taskConfigurationDtoService.deleteByPrimaryKey(taskConfigurationDto.getId());
        // remove job from zk
        removeTaskFromZooKeeper(taskId, zooKeeper);
    }

    /**
     * 删除zk上的作业结点。先持久化config/toDelete结点，让executor收到该事件，shutdown自身的该作业。如果所有executor都已经shutdown该作业，则才可以安全删除作业结点。
     *
     * @return 等待executor shutdown作业，等待一定时间后，如果executor还没完全shutdown，则放弃等待，返回false。
     * 否则，在等待时间内，executor都shutdown完全，则删除作业结点，并返回true。
     */
    private boolean removeTaskFromZooKeeper(String taskId, ZooKeeper zooKeeper) throws RuntimeException {
        // 1.作业的executor全online的情况，添加toDelete节点，触发监听器动态删除节点
        String toDeleteNodePath = TaskNodePath.getConfigNodePath(taskId, "toDelete");
        if (zooKeeper.checkExists(toDeleteNodePath)) {
            zooKeeper.deleteRecursive(toDeleteNodePath);
        }
        zooKeeper.create(toDeleteNodePath);
        //
        for (int i = 0; i < 20; i++) {
            // 2.作业的executor全offline的情况，或有几个online，几个offline的情况
            String executorsNodePath = TaskNodePath.getExecutorsNodePath(taskId);
            if (!zooKeeper.checkExists(executorsNodePath)) {
                // (1)如果不存在$Job/JobName/servers节点，说明该作业没有任何executor接管，可直接删除作业节点
                zooKeeper.deleteRecursive(TaskNodePath.getTaskNodePath(taskId));
                return true;
            }
            // (2)如果该作业servers下没有任何executor，可直接删除作业节点
            List<String> executorList = zooKeeper.getChildren(executorsNodePath);
            if (CollectionUtils.isEmpty(executorList)) {
                zooKeeper.deleteRecursive(TaskNodePath.getTaskNodePath(taskId));
                return true;
            }
            // (3)只要该作业没有一个能运行的该作业的executor在线，那么直接删除作业节点
            boolean hasOnlineExecutor = false;
            for (String executor : executorList) {
                if (zooKeeper.checkExists(ExecutorNodePath.getExecutorNodePath(executor, "ip")) && zooKeeper.checkExists(TaskNodePath.getExecutorStatus(taskId, executor))) {
                    hasOnlineExecutor = true;
                } else {
                    zooKeeper.deleteRecursive(TaskNodePath.getExecutorsNodePath(taskId, executor));
                }
            }
            if (!hasOnlineExecutor) {
                zooKeeper.deleteRecursive(TaskNodePath.getTaskNodePath(taskId));
                return true;
            }
            try {
                Thread.sleep(1000);
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }

        return false;

    }

    @Override
    public List<ExecutorProvided> getExecutorProvidedList(String nameSpace, String taskId) throws RuntimeException {
        TaskConfigurationDto taskConfigurationDto = taskConfigurationDtoService.getTaskConfigurationDtoByNameSpaceAndTaskId(nameSpace, taskId);
        if (taskConfigurationDto == null) {
            throw new RuntimeException("不能获取该作业（" + taskId + "）可选择的优先Executor，因为该作业不存在");
        }
        ZooKeeper zooKeeper = zooKeeperService.getZooKeeper(nameSpace);
        List<ExecutorProvided> executorProvidedList = new ArrayList<>();
        String executorsNodePath = Node.getExecutorsNodePath();
        if (!zooKeeper.checkExists(executorsNodePath)) {
            return executorProvidedList;
        }
        List<String> executorIdList = zooKeeper.getChildren(executorsNodePath);
        if (executorIdList == null) {
            executorIdList = new ArrayList<>();
        }
        if (!executorIdList.isEmpty()) {
            for (String executorId : executorIdList) {
                if (!executorId.startsWith("@")) {
                    ExecutorProvided executorProvided = new ExecutorProvided();
                    executorProvided.setExecutorType(ExecutorType.PHYSICAL_MACHINE);
                    executorProvided.setExecutorId(executorId);
                    executorProvided.setLostTraffic(zooKeeper.checkExists(Node.getExecutorLostTrafficNodePath(executorId)));
                    String ip = zooKeeper.getData(Node.getExecutorIpNodePath(executorId));
                    if (StringUtils.isNotBlank(ip)) {
                        executorProvided.setExecutorProvidedStatus(ExecutorProvidedStatus.ONLINE);
                        executorProvided.setIp(ip);
                    } else {
                        executorProvided.setExecutorProvidedStatus(ExecutorProvidedStatus.OFFLINE);
                    }
                    executorProvidedList.add(executorProvided);
                }

            }
            for (String executorId : executorIdList) {
                if (executorId.startsWith("@")) {
                    ExecutorProvided executorProvided = new ExecutorProvided();
                    executorProvided.setExecutorType(ExecutorType.CONTAINER);
                    executorProvided.setExecutorId(executorId);
                    executorProvided.setLostTraffic(zooKeeper.checkExists(Node.getExecutorLostTrafficNodePath(executorId)));
                    String ip = zooKeeper.getData(Node.getExecutorIpNodePath(executorId));
                    if (StringUtils.isNotBlank(ip)) {
                        executorProvided.setExecutorProvidedStatus(ExecutorProvidedStatus.ONLINE);
                        executorProvided.setIp(ip);
                    } else {
                        executorProvided.setExecutorProvidedStatus(ExecutorProvidedStatus.OFFLINE);
                    }
                    executorProvidedList.add(executorProvided);
                }
            }
        }
        String preferExecutorIdListNodePath = TaskNodePath.getConfigNodePath(taskId, CONFIG_ITEM_PREFER_LIST);
        if (!zooKeeper.checkExists(preferExecutorIdListNodePath)) {
            return executorProvidedList;
        }
        String preferExecutorIdListSting = zooKeeper.getData(preferExecutorIdListNodePath);
        if (Strings.isNullOrEmpty(preferExecutorIdListSting)) {
            return executorProvidedList;
        }
        fillUpInPreferExecutorIdListList(zooKeeper, executorIdList, executorProvidedList, preferExecutorIdListSting);
        return executorProvidedList;
    }

    private void fillUpInPreferExecutorIdListList(ZooKeeper zooKeeper, List<String> executorIdList, List<ExecutorProvided> executorProvidedList, String preferExecutorIdListSting) {
        String[] preferExecutorIdList = preferExecutorIdListSting.split(",");
        for (String preferExecutorId : preferExecutorIdList) {
            if (!preferExecutorId.startsWith("@")) {
                boolean include = false;
                for (ExecutorProvided executorProvided : executorProvidedList) {
                    if (executorProvided.getExecutorId().equals(preferExecutorId)) {
                        include = true;
                        break;
                    }
                }
                if (!include) {
                    ExecutorProvided executorProvided = new ExecutorProvided();
                    executorProvided.setExecutorId(preferExecutorId);
                    executorProvided.setExecutorType(ExecutorType.PHYSICAL_MACHINE);
                    executorProvided.setExecutorProvidedStatus(ExecutorProvidedStatus.DELETED);
                    executorProvidedList.add(executorProvided);
                }
            } else {
                boolean include = false;
                for (ExecutorProvided executorProvided : executorProvidedList) {
                    if (executorProvided.getExecutorId().equals(preferExecutorId)) {
                        include = true;
                        break;
                    }
                }
                if (!include) {
                    ExecutorProvided executorProvided = new ExecutorProvided();
                    executorProvided.setExecutorId(preferExecutorId);
                    executorProvided.setExecutorType(ExecutorType.CONTAINER);
                    executorProvided.setExecutorProvidedStatus(ExecutorProvidedStatus.DELETED);
                    executorProvidedList.add(executorProvided);
                }
            }
        }
    }

    @Transactional(rollbackFor = Exception.class)
    @Override
    public void setPreferList(String namespace, String taskId, String preferList, String updatedBy)
            throws RuntimeException {
        // save to db
        TaskConfigurationDto oldTaskConfigurationDto = taskConfigurationDtoService.getTaskConfigurationDtoByNameSpaceAndTaskId(namespace, taskId);
        if (oldTaskConfigurationDto == null) {
            throw new RuntimeException("设置该作业（" + taskId + "）优先Executor失败，因为该作业不存在");
        }

        TaskConfigurationDto newTaskConfigurationDto = new TaskConfigurationDto();
        org.springframework.beans.BeanUtils.copyProperties(oldTaskConfigurationDto, newTaskConfigurationDto);
        newTaskConfigurationDto.setPreferList(preferList);
        taskConfigurationDtoService.updateNewAndSaveOld2History(newTaskConfigurationDto, oldTaskConfigurationDto, updatedBy);

        // save to zk
        ZooKeeper zooKeeper = zooKeeperService.getZooKeeper(namespace);
        String taskConfigPreferListNodePath = Node.getTaskConfigPreferListNodePath(taskId);
        zooKeeper.update(taskConfigPreferListNodePath, preferList);
        // delete and create the forceShard node
        String taskConfigForceShardNodePath = Node.getTaskConfigForceShardNodePath(taskId);
        zooKeeper.delete(taskConfigForceShardNodePath);
        zooKeeper.create(taskConfigForceShardNodePath);
    }


    private void validateTaskConfiguration(TaskConfiguration taskConfiguration) throws RuntimeException {
        // 作业名必填
        String taskId = taskConfiguration.getTaskId();
        if (StringUtils.isBlank(taskId)) {
            throw new RuntimeException("作业名必填");
        }
        // 作业名只允许包含：数字0-9、小写字符a-z、大写字符A-Z、下划线_
        if (!taskId.matches("[0-9a-zA-Z_]*")) {
            throw new RuntimeException("作业名只允许包含：数字0-9、小写字符a-z、大写字符A-Z、下划线_");
        }

        // 作业类型必填
        if (StringUtils.isBlank(taskConfiguration.getTaskType())) {
            throw new RuntimeException("作业类型必填");
        }
        // 验证作业类型
        TaskType taskType = TaskType.getJobType(taskConfiguration.getTaskType());
        if (taskType == TaskType.UNKNOWN_JOB) {
            throw new RuntimeException("作业类型未知");
        }
        // 如果是JAVA作业，作业实现类必填
        if (StringUtils.isBlank(taskConfiguration.getTaskClass())) {
            throw new RuntimeException("对于java作业，作业实现类必填");
        }
        // 校验cron
        validateCronFieldOfJobConfig(taskConfiguration);
        // 校验shardingItemParameters
        validateShardingItemFieldOfTaskConfiguration(taskConfiguration);


    }


    private void validateCronFieldOfJobConfig(TaskConfiguration taskConfiguration) throws RuntimeException {
        if (TaskType.isCron(TaskType.getJobType(taskConfiguration.getTaskType()))) {
            // cron表达式必填
            if (taskConfiguration.getCronExpression() == null || taskConfiguration.getCronExpression().trim().isEmpty()) {
                throw new RuntimeException("对于cron作业，cron表达式必填");
            }
            // cron表达式语法验证
            try {
                CronExpression.validateExpression(taskConfiguration.getCronExpression());
            } catch (ParseException e) {
                throw new RuntimeException("cron表达式语法有误" + e);
            }
        } else {
            taskConfiguration.setCronExpression(""); // 其他类型的不需要持久化保存cron表达式
        }
    }

    private void validateShardingItemFieldOfTaskConfiguration(TaskConfiguration taskConfiguration) throws RuntimeException {
        {
            // 分片参数不能小于分片总数
            if (taskConfiguration.getTaskItemTotalCount() == null || taskConfiguration.getTaskItemTotalCount() < 1) {
                throw new RuntimeException("分片数不能为空，并且不能小于1");
            }
            if ((taskConfiguration.getTaskItemTotalCount() > 0) && (taskConfiguration.getTaskItemParameters() == null
                    || taskConfiguration.getTaskItemParameters().trim().isEmpty()
                    || taskConfiguration.getTaskItemParameters().split(",").length < taskConfiguration.getTaskItemTotalCount())) {
                throw new RuntimeException("分片参数不能小于分片总数");
            }
            validateShardingItemFormat(taskConfiguration);
        }
    }

    private void validateShardingItemFormat(TaskConfiguration taskConfiguration) throws RuntimeException {
        String shardingItemParameters = taskConfiguration.getTaskItemParameters();
        String[] shardingItemParameterss = shardingItemParameters.trim().split(",");
        for (int i = 0; i < shardingItemParameterss.length; i++) {
            String shardingItemParameter = shardingItemParameterss[i];
            if (!shardingItemParameter.contains("=")) {
                throw new RuntimeException(String.format("分片参数'%s'格式有误", shardingItemParameter));
            }
            String key = shardingItemParameter.trim().split("=")[0].trim();
            boolean isNumeric = StringUtils.isNumeric(key);
            if (!isNumeric) {
                throw new RuntimeException(String.format("分片参数'%s'格式有误", taskConfiguration.getTaskItemParameters()));
            }
        }
    }

    @Transactional(rollbackFor = Exception.class)
    @Override
    public void addJob(String nameSpace, TaskConfiguration taskConfiguration, String createdBy) throws RuntimeException {
        validateTaskConfiguration(taskConfiguration);
        // 如果数据存在相同作业名，则抛异常
        // 直接再查一次，不使用unSystemJobs，因为也不能与系统作业名相同
        String taskId = taskConfiguration.getTaskId();
        if (taskConfigurationDtoService.getTaskConfigurationDtoByNameSpaceAndTaskId(nameSpace, taskId) != null) {
            throw new RuntimeException(String.format("该作业(%s)已经存在", taskId));
        }
        // 如果zk存在该作业，则尝试删除
        ZooKeeper zooKeeper = zooKeeperService.getZooKeeper(nameSpace);
        if (zooKeeper.checkExists(TaskNodePath.getTaskNodePath(taskId))) {
            if (!removeTaskFromZooKeeper(taskId, zooKeeper)) {
                throw new RuntimeException(String.format("该作业(%s)正在删除中，请稍后再试", taskId));
            }
        }
        // 该域作业总数不能超过一定数量
        int maxTaskNum = getMaxTaskNum();
        if (taskIncExceeds(nameSpace, maxTaskNum, 1)) {
            throw new RuntimeException(String.format("总作业数超过最大限制(%d)，作业名%s创建失败", maxTaskNum, taskId));
        }
        // 如果是copy作业，则从数据库中复制被拷贝的作业的配置到新的作业配置
        // 设置作业配置字段默认值，并且强制纠正某些字段
        correctConfigValueWhenAddTask(taskConfiguration);
        // 添加该作业到数据库
        taskConfigurationDtoService.create(constructTaskConfigurationDto(nameSpace, taskConfiguration, createdBy, createdBy));

        // 添加该作业配置到zk，并联动更新关联作业的上下游
        createTaskConfigurationToZooKeeper(taskConfiguration, zooKeeper);
    }

    private TaskConfigurationDto constructTaskConfigurationDto(String nameSpace, TaskConfiguration taskConfiguration, String createdBy, String updatedBy) {
        TaskConfigurationDto taskConfigurationDto = new TaskConfigurationDto();
        BeanUtils.copyProperties(taskConfiguration, taskConfigurationDto);
        Date now = new Date();
        if (StringUtils.isNotBlank(createdBy)) {
            taskConfigurationDto.setCreateTime(now);
            taskConfigurationDto.setCreateBy(createdBy);
        }
        taskConfigurationDto.setLastUpdateTime(now);
        taskConfigurationDto.setLastUpdateBy(updatedBy);
        taskConfigurationDto.setNameSpace(nameSpace);
        return taskConfigurationDto;
    }

    private void correctConfigValueWhenAddTask(TaskConfiguration taskConfiguration) {
        taskConfiguration.setDefaultValues();
        taskConfiguration.setEnabled(false);
        TaskType taskType = TaskType.getJobType(taskConfiguration.getTaskType());

        if (TaskType.isPassive(taskType)) {
            taskConfiguration.setRerun(false);
        }


    }

    @Override
    public int getMaxTaskNum() {
        int DEFAULT_MAX_TASK_NUM = 128;
        return DEFAULT_MAX_TASK_NUM;
    }

    private int getMaxZooKeeperNodeDataLength() {
        // 最大允许显示的job log为zk默认的max jute buffer size
        int DEFAULT_MAX_ZOOKEEPER_NODE_DATA_LENGTH = 1048576;
        return DEFAULT_MAX_ZOOKEEPER_NODE_DATA_LENGTH;
    }

    @Override
    public boolean taskIncExceeds(String nameSpace, int maxJobNum, int inc) throws RuntimeException {
        if (maxJobNum <= 0) {
            return false;
        }
        int taskSize = getTaskConfigurationList(nameSpace).size();
        return (taskSize + inc) > maxJobNum;
    }

    @Override
    public List<TaskConfiguration> getTaskConfigurationList(String nameSpace) throws RuntimeException {
        List<TaskConfiguration> taskConfigurationList = new ArrayList<>();
        List<TaskConfigurationDto> taskConfiguration4DBList = taskConfigurationDtoService.getTaskConfigurationDtoListByNamespace(nameSpace);
        if (taskConfiguration4DBList != null) {
            for (TaskConfigurationDto taskConfigurationDto : taskConfiguration4DBList) {
                TaskConfiguration taskConfiguration = new TaskConfiguration();
                BeanUtils.copyProperties(taskConfigurationDto, taskConfiguration);
                taskConfigurationList.add(taskConfiguration);
            }
        }
        return taskConfigurationList;
    }


    @Override
    public List<TaskConfigurationDto> getUnSystemJobsWithCondition(String namespace, Map<String, Object> condition, int page, int size) throws RuntimeException {
        List<TaskConfigurationDto> taskConfiguration4DBList = getJobConfigByStatusWithCondition(namespace, condition, page, size);
        if (CollectionUtils.isEmpty(taskConfiguration4DBList)) {
            return new ArrayList<>();
        }
        return taskConfiguration4DBList;
    }

    private List<TaskConfigurationDto> getJobConfigByStatusWithCondition(String nameSpace, Map<String, Object> condition, int page, int size) throws RuntimeException {
        TaskStatus taskStatus = (TaskStatus) condition.get("jobStatus");
        if (taskStatus == null) {
            return taskConfigurationDtoService.getTaskConfigurationDtoListByNamespaceWithCondition(nameSpace, condition, PageableUtil.generatePageble(page, size));
        }

        List<TaskConfigurationDto> taskConfiguration4DBList = new ArrayList<TaskConfigurationDto>();
        List<TaskConfigurationDto> enabledTaskConfiguration4DBList = taskConfigurationDtoService.getTaskConfigurationDtoListByNamespaceWithCondition(nameSpace, condition, null);
        for (TaskConfigurationDto taskConfiguration4DB : enabledTaskConfiguration4DBList) {
            TaskStatus taskStatus1 = getTaskStatus(nameSpace, taskConfiguration4DB.getTaskId());
            if (taskStatus.equals(taskStatus1)) {
                taskConfiguration4DBList.add(taskConfiguration4DB);
            }
        }
        return taskConfiguration4DBList;
    }

    @Override
    public int countUnSystemJobsWithCondition(String nameSpace, Map<String, Object> condition) throws RuntimeException {
        return taskConfigurationDtoService.countTaskConfigurationDtoByNamespaceWithCondition(nameSpace, condition);
    }

    @Override
    public int countEnabledUnSystemJobs(String nameSpace) throws RuntimeException {
        return taskConfigurationDtoService.countEnabledUnSystemJobsByNamespace(nameSpace);
    }

    @Override
    public List<String> getUnSystemJobNames(String nameSpace) throws RuntimeException {
        List<String> taskIdList = new ArrayList<>();
        List<TaskConfigurationDto> taskConfiguration4DBList = taskConfigurationDtoService.getTaskConfigurationDtoListByNamespace(nameSpace);
        if (taskConfiguration4DBList != null) {
            for (TaskConfigurationDto taskConfiguration4DB : taskConfiguration4DBList) {
                taskIdList.add(taskConfiguration4DB.getTaskId());
            }
        }
        return taskIdList;
    }

    @Override
    public List<String> getTaskIdList(String nameSpace) throws RuntimeException {
        List<String> taskIdList = taskConfigurationDtoService.findConfigNamesByNameSpace(nameSpace);
        return taskIdList != null ? taskIdList : Lists.<String>newArrayList();
    }

    @Override
    public void saveTaskConfigurationToZooKeeperWithSetDefaultValues(String nameSpace, TaskConfiguration taskConfiguration) throws RuntimeException {
        taskConfiguration.setDefaultValues();
        ZooKeeper zooKeeper = zooKeeperService.getZooKeeper(nameSpace);
        saveTaskConfigurationToZooKeeper(taskConfiguration, zooKeeper);
    }

    @Override
    public void saveTaskConfigurationToZooKeeperWithSetDefaultValues(TaskConfiguration taskConfiguration, ZooKeeper zooKeeper) {
        taskConfiguration.setDefaultValues();
        saveTaskConfigurationToZooKeeper(taskConfiguration, zooKeeper);
    }

    private void createTaskConfigurationToZooKeeper(TaskConfiguration taskConfiguration, ZooKeeper zooKeeper) throws RuntimeException {
        try {
            String taskName = taskConfiguration.getTaskId();
            // 添加作业根节点和config结点
            zooKeeper.create(TaskNodePath.getConfigNodePath(taskName), "");
            TransactionalZooKeeper transactionalZooKeeper = zooKeeper.useTransaction();

            // 添加作业
            transactionalZooKeeper
                    .create(TaskNodePath.getConfigNodePath(taskName, CONFIG_ITEM_ENABLED), taskConfiguration.getEnabled())
                    .create(TaskNodePath.getConfigNodePath(taskName, CONFIG_ITEM_DESCRIPTION), taskConfiguration.getDescription())
                    .create(TaskNodePath.getConfigNodePath(taskName, CONFIG_ITEM_JOB_TYPE), taskConfiguration.getTaskType())

                    .create(TaskNodePath.getConfigNodePath(taskName, CONFIG_ITEM_SHARDING_ITEM_PARAMETERS), taskConfiguration.getTaskItemParameters())
                    .create(TaskNodePath.getConfigNodePath(taskName, CONFIG_ITEM_JOB_PARAMETER), taskConfiguration.getTaskParameter())

                    .create(TaskNodePath.getConfigNodePath(taskName, CONFIG_ITEM_FAILOVER), taskConfiguration.getFailover())
                    .create(TaskNodePath.getConfigNodePath(taskName, CONFIG_ITEM_MONITOR_EXECUTION), "true")
                    .create(TaskNodePath.getConfigNodePath(taskName, CONFIG_ITEM_TIMEOUT_4_ALARM_SECONDS), taskConfiguration.getTimeout4AlarmSeconds())
                    .create(TaskNodePath.getConfigNodePath(taskName, CONFIG_ITEM_TIMEOUT_SECONDS), taskConfiguration.getTimeoutSeconds())

                    .create(TaskNodePath.getConfigNodePath(taskName, CONFIG_ITEM_CRON), taskConfiguration.getCronExpression())
                    .create(TaskNodePath.getConfigNodePath(taskName, CONFIG_ITEM_PAUSE_PERIOD_DATE), taskConfiguration.getPauseDatePeriod())
                    .create(TaskNodePath.getConfigNodePath(taskName, CONFIG_ITEM_PAUSE_PERIOD_TIME), taskConfiguration.getPauseTimePeriod())
                    .create(TaskNodePath.getConfigNodePath(taskName, CONFIG_ITEM_PROCESS_COUNT_INTERVAL_SECONDS), taskConfiguration.getProcessCountIntervalSeconds())
                    .create(TaskNodePath.getConfigNodePath(taskName, CONFIG_ITEM_SHARDING_TOTAL_COUNT), taskConfiguration.getTaskItemTotalCount())
                    .create(TaskNodePath.getConfigNodePath(taskName, CONFIG_ITEM_SHOW_NORMAL_LOG), taskConfiguration.getShowNormalLog())
                    .create(TaskNodePath.getConfigNodePath(taskName, CONFIG_ITEM_LOAD_LEVEL), taskConfiguration.getLoadLevel())
                    .create(TaskNodePath.getConfigNodePath(taskName, CONFIG_ITEM_PREFER_LIST), taskConfiguration.getPreferList())
                    .create(TaskNodePath.getConfigNodePath(taskName, CONFIG_ITEM_USE_DISPREFER_LIST), taskConfiguration.getUseDisPreferList())


                    .create(TaskNodePath.getConfigNodePath(taskName, CONFIG_ITEM_RERUN), taskConfiguration.getRerun())

                    .create(TaskNodePath.getConfigNodePath(taskName, CONFIG_ITEM_JOB_CLASS), taskConfiguration.getTaskClass());
            // 注意！！！ jobClass要最后更新，因为executor认为该结点为作业添加完成

            // 提交事务
            transactionalZooKeeper.commit();
        } catch (Exception e) {
            log.error("create job to zk failed", e);
            throw new RuntimeException(e);
        }
    }

    private void saveTaskConfigurationToZooKeeper(TaskConfiguration taskConfiguration, ZooKeeper zooKeeper) {
        String jobName = taskConfiguration.getTaskId();
        zooKeeper.createIfNotExist(TaskNodePath.getConfigNodePath(jobName, CONFIG_ITEM_ENABLED), taskConfiguration.getEnabled());
        zooKeeper.createIfNotExist(TaskNodePath.getConfigNodePath(jobName, CONFIG_ITEM_DESCRIPTION), taskConfiguration.getDescription());
        zooKeeper.createIfNotExist(TaskNodePath.getConfigNodePath(jobName, CONFIG_ITEM_JOB_TYPE), taskConfiguration.getTaskType());

        zooKeeper.createIfNotExist(TaskNodePath.getConfigNodePath(jobName, CONFIG_ITEM_SHARDING_ITEM_PARAMETERS), taskConfiguration.getTaskItemParameters());
        zooKeeper.createIfNotExist(TaskNodePath.getConfigNodePath(jobName, CONFIG_ITEM_JOB_PARAMETER), taskConfiguration.getTaskParameter());

        zooKeeper.createIfNotExist(TaskNodePath.getConfigNodePath(jobName, CONFIG_ITEM_FAILOVER), taskConfiguration.getFailover());
        zooKeeper.createIfNotExist(TaskNodePath.getConfigNodePath(jobName, CONFIG_ITEM_MONITOR_EXECUTION), "true");
        zooKeeper.createIfNotExist(TaskNodePath.getConfigNodePath(jobName, CONFIG_ITEM_TIMEOUT_4_ALARM_SECONDS), taskConfiguration.getTimeout4AlarmSeconds());
        zooKeeper.createIfNotExist(TaskNodePath.getConfigNodePath(jobName, CONFIG_ITEM_TIMEOUT_SECONDS), taskConfiguration.getTimeoutSeconds());
        zooKeeper.createIfNotExist(TaskNodePath.getConfigNodePath(jobName, CONFIG_ITEM_CRON), taskConfiguration.getCronExpression());
        zooKeeper.createIfNotExist(TaskNodePath.getConfigNodePath(jobName, CONFIG_ITEM_PAUSE_PERIOD_DATE), taskConfiguration.getPauseDatePeriod());
        zooKeeper.createIfNotExist(TaskNodePath.getConfigNodePath(jobName, CONFIG_ITEM_PAUSE_PERIOD_TIME), taskConfiguration.getPauseTimePeriod());
        zooKeeper.createIfNotExist(TaskNodePath.getConfigNodePath(jobName, CONFIG_ITEM_PROCESS_COUNT_INTERVAL_SECONDS), taskConfiguration.getProcessCountIntervalSeconds());
        zooKeeper.createIfNotExist(TaskNodePath.getConfigNodePath(jobName, CONFIG_ITEM_SHARDING_TOTAL_COUNT), taskConfiguration.getTaskItemTotalCount());
        zooKeeper.createIfNotExist(TaskNodePath.getConfigNodePath(jobName, CONFIG_ITEM_SHOW_NORMAL_LOG), taskConfiguration.getShowNormalLog());
        zooKeeper.createIfNotExist(TaskNodePath.getConfigNodePath(jobName, CONFIG_ITEM_LOAD_LEVEL), taskConfiguration.getLoadLevel());
        zooKeeper.createIfNotExist(TaskNodePath.getConfigNodePath(jobName, CONFIG_ITEM_PREFER_LIST), taskConfiguration.getPreferList());
        zooKeeper.createIfNotExist(TaskNodePath.getConfigNodePath(jobName, CONFIG_ITEM_USE_DISPREFER_LIST), taskConfiguration.getUseDisPreferList());


        zooKeeper.createIfNotExist(TaskNodePath.getConfigNodePath(jobName, CONFIG_ITEM_RERUN), taskConfiguration.getRerun());

        zooKeeper.createIfNotExist(TaskNodePath.getConfigNodePath(jobName, CONFIG_ITEM_JOB_CLASS), taskConfiguration.getTaskClass());
        // 注意！！！ jobClass要最后更新，因为executor认为该结点为作业添加完成
    }

    @Override
    public List<TaskResult> importTasks(String nameSpace, MultipartFile file, String createdBy)
            throws RuntimeException {
        try {
            Workbook workbook = Workbook.getWorkbook(file.getInputStream());

            Sheet[] sheets = workbook.getSheets();
            List<TaskConfiguration> jobConfigList = new ArrayList<>();
            // 第一行为配置项提示，从第二行开始为作业配置信息
            // 先获取数据并检测内容格式的正确性
            for (int i = 0; i < sheets.length; i++) {
                Sheet sheet = sheets[i];
                int rows = sheet.getRows();
                for (int row = 1; row < rows; row++) {
                    Cell[] rowCells = sheet.getRow(row);
                    // 如果这一行的表格全为空，则跳过这一行。
                    if (!isBlankRow(rowCells)) {
                        jobConfigList.add(convertTaskConfiguration(i + 1, row + 1, rowCells));
                    }
                }
            }
            int maxJobNum = getMaxTaskNum();
            if (taskIncExceeds(nameSpace, maxJobNum, jobConfigList.size())) {
                throw new RuntimeException(
                        String.format("总作业数超过最大限制(%d)，导入失败", maxJobNum));
            }
            return doCreateTaskFromImportTaskFile(nameSpace, jobConfigList, createdBy);
        } catch (RuntimeException e) {
            throw e;
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    protected List<TaskResult> doCreateTaskFromImportTaskFile(String nameSpace, List<TaskConfiguration> taskConfigurationList, String createdBy) throws RuntimeException {
        Map<String, TaskResult> resultMap = new LinkedHashMap<>();

        for (TaskConfiguration taskConfiguration : taskConfigurationList) {
            String taskId = taskConfiguration.getTaskId();
            TaskResult taskResult = new TaskResult();
            taskResult.setTaskId(taskId);
            try {

                addJob(nameSpace, taskConfiguration, createdBy);
                taskResult.setSuccess(true);

            } catch (Exception e) {
                taskResult.setSuccess(false);
                taskResult.setMessage(e.toString());
                log.warn(e.getMessage(), e);
            }
            resultMap.put(taskId, taskResult);
        }

        return new ArrayList<>(resultMap.values());
    }

    private boolean isBlankRow(Cell[] rowCells) {
        for (int i = 0; i < rowCells.length; i++) {
            if (!CellType.EMPTY.equals(rowCells[i].getType())) {
                return false;
            }
        }
        return true;
    }

    private TaskConfiguration convertTaskConfiguration(int sheetNumber, int rowNumber, Cell[] rowCells)
            throws RuntimeException {
        String jobName = getContents(rowCells, 0);
        if (jobName == null || jobName.trim().isEmpty()) {
            throw new RuntimeException(
                    createExceptionMessage(sheetNumber, rowNumber, 1, "作业名必填。"));
        }
        if (!jobName.matches("[0-9a-zA-Z_]*")) {
            throw new RuntimeException(
                    createExceptionMessage(sheetNumber, rowNumber, 1, "作业名只允许包含：数字0-9、小写字符a-z、大写字符A-Z、下划线_。"));
        }
        TaskConfiguration taskConfiguration = new TaskConfiguration();
        taskConfiguration.setTaskId(jobName);

        String jobType = getContents(rowCells, 1);
        if (jobType == null || jobType.trim().isEmpty()) {
            throw new RuntimeException(
                    createExceptionMessage(sheetNumber, rowNumber, 2, "作业类型必填。"));
        }
        TaskType taskType = TaskType.getJobType(jobType);
        if (taskType == TaskType.UNKNOWN_JOB) {
            throw new RuntimeException(
                    createExceptionMessage(sheetNumber, rowNumber, 2, "作业类型未知。"));
        }
        taskConfiguration.setTaskType(jobType);

        String jobClass = getContents(rowCells, 2);
        if ((jobClass == null || jobClass.trim().isEmpty())) {
            throw new RuntimeException(
                    createExceptionMessage(sheetNumber, rowNumber, 3, "对于java作业，作业实现类必填。"));
        }
        taskConfiguration.setTaskClass(jobClass);

        String cron = getContents(rowCells, 3);
        if (TaskType.isCron(taskType)) {
            if (cron == null || cron.trim().isEmpty()) {
                throw new RuntimeException(
                        createExceptionMessage(sheetNumber, rowNumber, 4, "对于cron作业，cron表达式必填。"));
            }
            cron = cron.trim();
            try {
                CronExpression.validateExpression(cron);
            } catch (ParseException e) {
                throw new RuntimeException(
                        createExceptionMessage(sheetNumber, rowNumber, 4, "cron表达式语法有误，" + e));
            }
        } else {
            cron = "";// 其他类型的不需要持久化保存cron表达式
        }

        taskConfiguration.setCronExpression(cron);

        taskConfiguration.setDescription(getContents(rowCells, 4));


        int shardingTotalCount = 1;
        {
            String tmp = getContents(rowCells, 6);
            if (tmp != null && !tmp.trim().isEmpty()) {
                try {
                    shardingTotalCount = Integer.parseInt(tmp);
                } catch (NumberFormatException e) {
                    throw new RuntimeException(
                            createExceptionMessage(sheetNumber, rowNumber, 7, "分片数有误，" + e));
                }
            } else {
                throw new RuntimeException(
                        createExceptionMessage(sheetNumber, rowNumber, 7, "分片数必填"));
            }
            if (shardingTotalCount < 1) {
                throw new RuntimeException(
                        createExceptionMessage(sheetNumber, rowNumber, 7, "分片数不能小于1"));
            }
            taskConfiguration.setTaskItemTotalCount(shardingTotalCount);
        }

        int timeoutSeconds = 0;
        try {
            String tmp = getContents(rowCells, 7);
            if (tmp != null && !tmp.trim().isEmpty()) {
                timeoutSeconds = Integer.parseInt(tmp.trim());
            }
        } catch (NumberFormatException e) {
            throw new RuntimeException(
                    createExceptionMessage(sheetNumber, rowNumber, 8, "超时（Kill线程/进程）时间有误，" + e));
        }
        taskConfiguration.setTimeoutSeconds(timeoutSeconds);

        taskConfiguration.setTaskParameter(getContents(rowCells, 8));

        String shardingItemParameters = getContents(rowCells, 9);
        if ((shardingTotalCount > 0) && (shardingItemParameters == null || shardingItemParameters.trim().isEmpty() || shardingItemParameters.split(",").length < shardingTotalCount)) {
            throw new RuntimeException(createExceptionMessage(sheetNumber, rowNumber, 10, "分片参数不能小于分片总数。"));
        }
        taskConfiguration.setTaskItemParameters(shardingItemParameters);


        taskConfiguration.setPreferList(getContents(rowCells, 12));
        taskConfiguration.setUseDisPreferList(!Boolean.parseBoolean(getContents(rowCells, 13)));

        int processCountIntervalSeconds = 300;
        try {
            String tmp = getContents(rowCells, 14);
            if (tmp != null && !tmp.trim().isEmpty()) {
                processCountIntervalSeconds = Integer.parseInt(tmp.trim());
            }
        } catch (NumberFormatException e) {
            throw new RuntimeException(
                    createExceptionMessage(sheetNumber, rowNumber, 15, "统计处理数据量的间隔秒数有误，" + e));
        }
        taskConfiguration.setProcessCountIntervalSeconds(processCountIntervalSeconds);

        int loadLevel = 1;
        try {
            String tmp = getContents(rowCells, 15);
            if (tmp != null && !tmp.trim().isEmpty()) {
                loadLevel = Integer.parseInt(tmp.trim());
            }
        } catch (NumberFormatException e) {
            throw new RuntimeException(
                    createExceptionMessage(sheetNumber, rowNumber, 16, "负荷有误，" + e));
        }
        taskConfiguration.setLoadLevel(loadLevel);

        taskConfiguration.setShowNormalLog(Boolean.valueOf(getContents(rowCells, 16)));

        taskConfiguration.setPauseDatePeriod(getContents(rowCells, 17));

        taskConfiguration.setPauseTimePeriod(getContents(rowCells, 18));


        String dependencies = getContents(rowCells, 23);
        if (dependencies != null && !dependencies.matches("[0-9a-zA-Z_,]*")) {
            throw new RuntimeException(createExceptionMessage(sheetNumber, rowNumber, 24, "依赖的作业只允许包含：数字0-9、小写字符a-z、大写字符A-Z、下划线_、英文逗号,"));
        }


        int timeout4AlarmSeconds = 0;
        try {
            String tmp = getContents(rowCells, 25);
            if (tmp != null && !tmp.trim().isEmpty()) {
                timeout4AlarmSeconds = Integer.parseInt(tmp.trim());
            }
        } catch (NumberFormatException e) {
            throw new RuntimeException(
                    createExceptionMessage(sheetNumber, rowNumber, 26, "超时（告警）时间有误，" + e));
        }
        taskConfiguration.setTimeout4AlarmSeconds(timeout4AlarmSeconds);

        String timeZone = getContents(rowCells, 26);
        if (timeZone == null || timeZone.trim().length() == 0) {
            timeZone = SaturnConstants.TIME_ZONE_ID_DEFAULT;
        } else {
            timeZone = timeZone.trim();
            if (!SaturnConstants.TIME_ZONE_IDS.contains(timeZone)) {
                throw new RuntimeException(createExceptionMessage(sheetNumber, rowNumber, 27, "时区有误"));
            }
        }


        Boolean failover = null;
        String failoverStr = getContents(rowCells, 27);
        if (StringUtils.isNotBlank(failoverStr)) {
            failover = Boolean.valueOf(failoverStr.trim());
            if (failover) {


                // 如果不上报运行状态，则强制设置为false
                // 上报运行状态失效，由算法决定是否上报，看下面setEnabledReport时的逻辑，看addJob
            }
        }
        taskConfiguration.setFailover(failover);

        Boolean rerun = null;
        String rerunStr = getContents(rowCells, 28);
        if (StringUtils.isNotBlank(rerunStr)) {
            rerun = Boolean.valueOf(rerunStr.trim());
            if (rerun) {

                if (TaskType.isPassive(taskType)) {
                    throw new RuntimeException(createExceptionMessage(sheetNumber, rowNumber, 29, "被动作业不支持rerun"));
                }
                // 如果不上报运行状态，则强制设置为false
                // 上报运行状态失效，由算法决定是否上报，看下面setEnabledReport时的逻辑，看addJob
            }
        }
        taskConfiguration.setRerun(rerun);


        int disableTimeoutSeconds = 0;
        try {
            String tmp = getContents(rowCells, 31);
            if (tmp != null && !tmp.trim().isEmpty()) {
                disableTimeoutSeconds = Integer.parseInt(tmp.trim());
            }
        } catch (NumberFormatException e) {
            throw new RuntimeException(
                    createExceptionMessage(sheetNumber, rowNumber, 32, "禁用超时（告警）时间有误，" + e));
        }


        return taskConfiguration;
    }

    private String getContents(Cell[] rowCell, int column) {
        if (rowCell.length > column) {
            return rowCell[column].getContents();
        }
        return null;
    }

    private String createExceptionMessage(int sheetNumber, int rowNumber, int columnNumber, String message) {
        return "内容格式有误，错误发生在表格页:" + sheetNumber + "，行号:" + rowNumber + "，列号:" + columnNumber + "，错误信息：" + message;
    }

    @Override
    public File exportJobs(String namespace) throws RuntimeException {
        try {
            File tmp = new File(SaturnConstants.CACHES_FILE_PATH,
                    "tmp_exportFile_" + System.currentTimeMillis() + "_" + random.nextInt(1000) + ".xls");
            if (!tmp.exists()) {
                //FileUtils.forceMkdir(tmp.getParentFile());
                tmp.createNewFile();
            }
            WritableWorkbook writableWorkbook = Workbook.createWorkbook(tmp);
            WritableSheet sheet1 = writableWorkbook.createSheet("Sheet1", 0);
            setExcelHeader(sheet1);
            List<TaskConfiguration> unSystemJobs = getTaskConfigurationList(namespace);
            // sort by jobName
            Collections.sort(unSystemJobs, new Comparator<TaskConfiguration>() {
                @Override
                public int compare(TaskConfiguration o1, TaskConfiguration o2) {
                    return o1.getTaskId().compareTo(o2.getTaskId());
                }
            });
            setExcelContent(namespace, sheet1, unSystemJobs);

            writableWorkbook.write();
            writableWorkbook.close();

            return tmp;
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public File exportSelectedJobs(String namespace, List<String> jobList) throws RuntimeException {
        try {
            File tmp = new File(SaturnConstants.CACHES_FILE_PATH,
                    "tmp_exportFile_" + System.currentTimeMillis() + "_" + random.nextInt(1000) + ".xls");
            if (!tmp.exists()) {
                //FileUtils.forceMkdir(tmp.getParentFile());
                tmp.createNewFile();
            }
            WritableWorkbook writableWorkbook = Workbook.createWorkbook(tmp);
            WritableSheet sheet1 = writableWorkbook.createSheet("Sheet1", 0);
            setExcelHeader(sheet1);
            // 单个域下作业量不会很大，直接复用之前的方法取该域下全部作业再过滤
            List<TaskConfiguration> taskConfigurationList = filterTaskConfigurationList(jobList, getTaskConfigurationList(namespace));
            // sort by jobName
            Collections.sort(taskConfigurationList, new Comparator<TaskConfiguration>() {
                @Override
                public int compare(TaskConfiguration o1, TaskConfiguration o2) {
                    return o1.getTaskId().compareTo(o2.getTaskId());
                }
            });
            setExcelContent(namespace, sheet1, taskConfigurationList);

            writableWorkbook.write();
            writableWorkbook.close();

            return tmp;
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    private List<TaskConfiguration> filterTaskConfigurationList(List<String> taskIdList, List<TaskConfiguration> taskConfigurationList1) {
        List<TaskConfiguration> taskConfigurationList = new ArrayList<>();
        for (TaskConfiguration taskConfiguration : taskConfigurationList1) {
            if (taskIdList.contains(taskConfiguration.getTaskId())) {
                taskConfigurationList.add(taskConfiguration);
            }
        }
        return taskConfigurationList;
    }

    protected void setExcelContent(String namespace, WritableSheet sheet1, List<TaskConfiguration> unSystemJobs)
            throws RuntimeException, WriteException {
        if (unSystemJobs != null && !unSystemJobs.isEmpty()) {
            ZooKeeper zooKeeper = zooKeeperService.getZooKeeper(namespace);
            for (int i = 0; i < unSystemJobs.size(); i++) {
                String jobName = unSystemJobs.get(i).getTaskId();
                sheet1.addCell(new Label(0, i + 1, jobName));
                sheet1.addCell(new Label(1, i + 1,
                        zooKeeper.getData(TaskNodePath.getConfigNodePath(jobName, CONFIG_ITEM_JOB_TYPE))));
                sheet1.addCell(new Label(2, i + 1,
                        zooKeeper.getData(TaskNodePath.getConfigNodePath(jobName, CONFIG_ITEM_JOB_CLASS))));
                sheet1.addCell(new Label(3, i + 1,
                        zooKeeper.getData(TaskNodePath.getConfigNodePath(jobName, CONFIG_ITEM_CRON))));
                sheet1.addCell(new Label(4, i + 1,
                        zooKeeper.getData(TaskNodePath.getConfigNodePath(jobName, CONFIG_ITEM_DESCRIPTION))));

                sheet1.addCell(new Label(6, i + 1, zooKeeper
                        .getData(TaskNodePath.getConfigNodePath(jobName, CONFIG_ITEM_SHARDING_TOTAL_COUNT))));
                sheet1.addCell(new Label(7, i + 1, zooKeeper
                        .getData(TaskNodePath.getConfigNodePath(jobName, CONFIG_ITEM_TIMEOUT_SECONDS))));
                sheet1.addCell(new Label(8, i + 1,
                        zooKeeper.getData(TaskNodePath.getConfigNodePath(jobName, CONFIG_ITEM_JOB_PARAMETER))));
                sheet1.addCell(new Label(9, i + 1, zooKeeper
                        .getData(TaskNodePath.getConfigNodePath(jobName, CONFIG_ITEM_SHARDING_ITEM_PARAMETERS))));

                sheet1.addCell(new Label(12, i + 1,
                        zooKeeper.getData(TaskNodePath.getConfigNodePath(jobName, CONFIG_ITEM_PREFER_LIST))));
                String useDispreferList = zooKeeper
                        .getData(TaskNodePath.getConfigNodePath(jobName, CONFIG_ITEM_USE_DISPREFER_LIST));
                if (useDispreferList != null) {
                    useDispreferList = String.valueOf(!Boolean.parseBoolean(useDispreferList));
                }
                sheet1.addCell(new Label(13, i + 1, useDispreferList));
                sheet1.addCell(new Label(14, i + 1, zooKeeper
                        .getData(TaskNodePath.getConfigNodePath(jobName, CONFIG_ITEM_PROCESS_COUNT_INTERVAL_SECONDS))));
                sheet1.addCell(new Label(15, i + 1,
                        zooKeeper.getData(TaskNodePath.getConfigNodePath(jobName, CONFIG_ITEM_LOAD_LEVEL))));
                sheet1.addCell(new Label(16, i + 1, zooKeeper
                        .getData(TaskNodePath.getConfigNodePath(jobName, CONFIG_ITEM_SHOW_NORMAL_LOG))));
                sheet1.addCell(new Label(17, i + 1, zooKeeper
                        .getData(TaskNodePath.getConfigNodePath(jobName, CONFIG_ITEM_PAUSE_PERIOD_DATE))));
                sheet1.addCell(new Label(18, i + 1, zooKeeper
                        .getData(TaskNodePath.getConfigNodePath(jobName, CONFIG_ITEM_PAUSE_PERIOD_TIME))));

                sheet1.addCell(new Label(20, i + 1,
                        zooKeeper.getData(TaskNodePath.getConfigNodePath(jobName, CONFIG_ITEM_JOB_DEGREE))));
                sheet1.addCell(new Label(21, i + 1, zooKeeper
                        .getData(TaskNodePath.getConfigNodePath(jobName, CONFIG_ITEM_ENABLED_REPORT))));


//				sheet1.addCell(new Label(24, i + 1,
//						curatorFrameworkOp.getData(JobNodePath.getConfigNodePath(jobName, CONFIG_ITEM_GROUPS))));
                // 批量分组后，分组只写到了db并没有写到zk，导出的Excel表格缺失分组名数据，所以这里直接取db的数据

                sheet1.addCell(new Label(25, i + 1, zooKeeper
                        .getData(TaskNodePath.getConfigNodePath(jobName, CONFIG_ITEM_TIMEOUT_4_ALARM_SECONDS))));
                sheet1.addCell(new Label(26, i + 1,
                        zooKeeper.getData(TaskNodePath.getConfigNodePath(jobName, CONFIG_ITEM_TIME_ZONE))));
                sheet1.addCell(new Label(27, i + 1,
                        zooKeeper.getData(TaskNodePath.getConfigNodePath(jobName, CONFIG_ITEM_FAILOVER))));
                sheet1.addCell(new Label(28, i + 1,
                        zooKeeper.getData(TaskNodePath.getConfigNodePath(jobName, CONFIG_ITEM_RERUN))));

                sheet1.addCell(new Label(31, i + 1,
                        zooKeeper.getData(TaskNodePath.getConfigNodePath(jobName, CONFIG_DISABLE_TIMEOUT_SECONDS))));
                sheet1.addCell(new Label(32, i + 1,
                        zooKeeper.getData(TaskNodePath.getConfigNodePath(jobName, CONFIG_ITEM_ENABLED))));
            }
        }
    }

    protected void setExcelHeader(WritableSheet sheet1) throws WriteException {
        sheet1.addCell(new Label(0, 0, "作业名称"));
        sheet1.addCell(new Label(1, 0, "作业类型"));
        sheet1.addCell(new Label(2, 0, "作业实现类"));
        sheet1.addCell(new Label(3, 0, "cron表达式"));
        sheet1.addCell(new Label(4, 0, "作业描述"));


        Label shardingTotalCountLabel = new Label(6, 0, "分片数");
        setCellComment(shardingTotalCountLabel, "对本地作业无效");
        sheet1.addCell(shardingTotalCountLabel);

        Label timeoutSecondsLabel = new Label(7, 0, "运行超时（Kill线程/进程）时间");
        setCellComment(timeoutSecondsLabel, "0表示无超时");
        sheet1.addCell(timeoutSecondsLabel);

        sheet1.addCell(new Label(8, 0, "自定义参数"));
        sheet1.addCell(new Label(9, 0, "分片序列号/参数对照表"));

        sheet1.addCell(new Label(11, 0, "执行结果发送的Channel"));

        Label preferListLabel = new Label(12, 0, "优先Executor");
        setCellComment(preferListLabel, "可填executorName，多个元素使用英文逗号隔开");
        sheet1.addCell(preferListLabel);

        Label usePreferListOnlyLabel = new Label(13, 0, "只使用优先Executor");
        setCellComment(usePreferListOnlyLabel, "默认为false");
        sheet1.addCell(usePreferListOnlyLabel);

        sheet1.addCell(new Label(14, 0, "统计处理数据量的间隔秒数"));
        sheet1.addCell(new Label(15, 0, "负荷"));
        sheet1.addCell(new Label(16, 0, "显示控制台输出日志"));
        sheet1.addCell(new Label(17, 0, "暂停日期段"));
        sheet1.addCell(new Label(18, 0, "暂停时间段"));

        Label useSerialLabel = new Label(19, 0, "串行消费");
        setCellComment(useSerialLabel, "默认为false");
        sheet1.addCell(useSerialLabel);

        Label jobDegreeLabel = new Label(20, 0, "作业重要等级");
        setCellComment(jobDegreeLabel, "0:没有定义,1:非线上业务,2:简单业务,3:一般业务,4:重要业务,5:核心业务");
        sheet1.addCell(jobDegreeLabel);

        Label enabledReportLabel = new Label(21, 0, "上报运行状态");
        setCellComment(enabledReportLabel, "对于定时作业，默认为true；对于消息作业，默认为false");
        sheet1.addCell(enabledReportLabel);

        Label jobModeLabel = new Label(22, 0, "作业模式");
        setCellComment(jobModeLabel, "用户不能添加系统作业");
        sheet1.addCell(jobModeLabel);

        Label dependenciesLabel = new Label(23, 0, "依赖的作业");
        setCellComment(dependenciesLabel, "作业的启用、禁用会检查依赖关系的作业的状态。依赖多个作业，使用英文逗号给开。该字段已过期。");
        sheet1.addCell(dependenciesLabel);


        Label timeout4AlarmSecondsLabel = new Label(25, 0, "运行超时（告警）时间");
        setCellComment(timeout4AlarmSecondsLabel, "0表示无超时");
        sheet1.addCell(timeout4AlarmSecondsLabel);

        Label timeZoneLabel = new Label(26, 0, "时区");
        setCellComment(timeZoneLabel, "作业运行时区");
        sheet1.addCell(timeZoneLabel);

        sheet1.addCell(new Label(27, 0, "failover"));

        sheet1.addCell(new Label(28, 0, "失败重跑"));


        Label disableTimeoutSecondsLabel = new Label(31, 0, "禁用超时（告警）时间");
        setCellComment(disableTimeoutSecondsLabel, "0表示无超时");
        sheet1.addCell(disableTimeoutSecondsLabel);

        sheet1.addCell(new Label(32, 0, "是否启用"));
    }

    protected void setCellComment(WritableCell cell, String comment) {
        WritableCellFeatures cellFeatures = new WritableCellFeatures();
        cellFeatures.setComment(comment);
        cell.setCellFeatures(cellFeatures);
    }


    @Override
    public TaskConfiguration getTaskConfigurationFromZooKeeper(String namespace, String taskId) throws RuntimeException {
        ZooKeeper zooKeeper = zooKeeperService.getZooKeeper(namespace);
        TaskConfiguration taskConfiguration = new TaskConfiguration();
        taskConfiguration.setTaskId(taskId);
        taskConfiguration.setTaskType(zooKeeper.getData(TaskNodePath.getConfigNodePath(taskId, CONFIG_ITEM_JOB_TYPE)));
        taskConfiguration.setTaskClass(zooKeeper.getData(TaskNodePath.getConfigNodePath(taskId, CONFIG_ITEM_JOB_CLASS)));
        // 兼容旧版没有msg_job。
        if (StringUtils.isBlank(taskConfiguration.getTaskType())) {
            if (taskConfiguration.getTaskClass().indexOf("script") >= 0) {
                taskConfiguration.setTaskType(TaskType.SHELL_JOB.name());
            } else {
                taskConfiguration.setTaskType(TaskType.JAVA_JOB.name());
            }
        }
        taskConfiguration.setTaskItemTotalCount(Integer.valueOf(zooKeeper.getData(TaskNodePath.getConfigNodePath(taskId, CONFIG_ITEM_SHARDING_TOTAL_COUNT))));

        taskConfiguration.setCronExpression(zooKeeper.getData(TaskNodePath.getConfigNodePath(taskId, CONFIG_ITEM_CRON)));
        taskConfiguration.setPauseDatePeriod(zooKeeper.getData(TaskNodePath.getConfigNodePath(taskId, CONFIG_ITEM_PAUSE_PERIOD_DATE)));
        taskConfiguration.setPauseTimePeriod(zooKeeper.getData(TaskNodePath.getConfigNodePath(taskId, CONFIG_ITEM_PAUSE_PERIOD_TIME)));
        taskConfiguration.setTaskItemParameters(zooKeeper.getData(TaskNodePath.getConfigNodePath(taskId, CONFIG_ITEM_SHARDING_ITEM_PARAMETERS)));
        taskConfiguration.setTaskParameter(zooKeeper.getData(TaskNodePath.getConfigNodePath(taskId, CONFIG_ITEM_JOB_PARAMETER)));
        taskConfiguration.setProcessCountIntervalSeconds(Integer.valueOf(zooKeeper.getData(TaskNodePath.getConfigNodePath(taskId, CONFIG_ITEM_PROCESS_COUNT_INTERVAL_SECONDS))));
        String timeout4AlarmSecondsStr = zooKeeper.getData(TaskNodePath.getConfigNodePath(taskId, CONFIG_ITEM_TIMEOUT_4_ALARM_SECONDS));
        if (Strings.isNullOrEmpty(timeout4AlarmSecondsStr)) {
            taskConfiguration.setTimeout4AlarmSeconds(0);
        } else {
            taskConfiguration.setTimeout4AlarmSeconds(Integer.valueOf(timeout4AlarmSecondsStr));
        }
        taskConfiguration.setTimeoutSeconds(Integer.valueOf(zooKeeper.getData(TaskNodePath.getConfigNodePath(taskId, CONFIG_ITEM_TIMEOUT_SECONDS))));
        String disableTimeoutSecondsStr = zooKeeper.getData(TaskNodePath.getConfigNodePath(taskId, CONFIG_DISABLE_TIMEOUT_SECONDS));

        String lv = zooKeeper.getData(TaskNodePath.getConfigNodePath(taskId, CONFIG_ITEM_LOAD_LEVEL));
        if (Strings.isNullOrEmpty(lv)) {
            taskConfiguration.setLoadLevel(1);
        } else {
            taskConfiguration.setLoadLevel(Integer.valueOf(lv));
        }

        taskConfiguration.setEnabled(Boolean.valueOf(zooKeeper.getData(TaskNodePath.getConfigNodePath(taskId, CONFIG_ITEM_ENABLED))));// 默认是禁用的
        taskConfiguration.setPreferList(zooKeeper.getData(TaskNodePath.getConfigNodePath(taskId, CONFIG_ITEM_PREFER_LIST)));
        String useDispreferList = zooKeeper.getData(TaskNodePath.getConfigNodePath(taskId, CONFIG_ITEM_USE_DISPREFER_LIST));
        if (Strings.isNullOrEmpty(useDispreferList)) {
            taskConfiguration.setUseDisPreferList(null);
        } else {
            taskConfiguration.setUseDisPreferList(Boolean.valueOf(useDispreferList));
        }


        taskConfiguration.setDescription(zooKeeper.getData(TaskNodePath.getConfigNodePath(taskId, CONFIG_ITEM_DESCRIPTION)));


        if (!zooKeeper.checkExists(TaskNodePath.getConfigNodePath(taskId, TaskServiceImpl.CONFIG_ITEM_SHOW_NORMAL_LOG))) {
            zooKeeper.create(TaskNodePath.getConfigNodePath(taskId, CONFIG_ITEM_SHOW_NORMAL_LOG));
        }


        taskConfiguration.setShowNormalLog(Boolean.valueOf(zooKeeper.getData(TaskNodePath.getConfigNodePath(taskId, CONFIG_ITEM_SHOW_NORMAL_LOG))));
        return taskConfiguration;
    }

    @Override
    public TaskConfiguration getTaskConfiguration(String nameSpace, String taskId) throws RuntimeException {
        TaskConfigurationDto taskConfiguration4DB = taskConfigurationDtoService.getTaskConfigurationDtoByNameSpaceAndTaskId(nameSpace, taskId);
        if (taskConfiguration4DB == null) {
            throw new RuntimeException(String.format("该作业(%s)不存在", taskId));
        }
        TaskConfiguration taskConfiguration = new TaskConfiguration();
        BeanUtils.copyProperties(taskConfiguration4DB, taskConfiguration);
        return taskConfiguration;
    }

    @Override
    public TaskStatus getTaskStatus(String nameSpace, String taskId) throws RuntimeException {
        TaskConfigurationDto taskConfiguration4DB = taskConfigurationDtoService.getTaskConfigurationDtoByNameSpaceAndTaskId(nameSpace, taskId);
        if (taskConfiguration4DB == null) {
            throw new RuntimeException("不能获取该作业（" + taskId + "）的状态，因为该作业不存在");
        }
        ZooKeeper zooKeeper = zooKeeperService.getZooKeeper(nameSpace);
        return getTaskStatus(taskId, zooKeeper, taskConfiguration4DB.getEnabled());
    }

    @Override
    public TaskStatus getTaskStatus(String nameSpace, TaskConfiguration taskConfiguration) throws RuntimeException {
        ZooKeeper zooKeeper = zooKeeperService.getZooKeeper(nameSpace);
        return getTaskStatus(taskConfiguration.getTaskId(), zooKeeper, taskConfiguration.getEnabled());
    }

    @Override
    public boolean isJobShardingAllocatedExecutor(String nameSpace, String taskId) throws RuntimeException {
        ZooKeeper zooKeeper = zooKeeperService.getZooKeeper(nameSpace);
        String executorsNodePath = TaskNodePath.getExecutorsNodePath(taskId);
        List<String> executorList = zooKeeper.getChildren(executorsNodePath);
        if (CollectionUtils.isEmpty(executorList)) {
            return false;
        }
        for (String executor : executorList) {
            String sharding = zooKeeper.getData(TaskNodePath.getExecutorsNodePath(taskId, executor, "sharding"));
            if (StringUtils.isNotBlank(sharding)) {
                return true;
            }
        }
        return false;
    }

    @Override
    public List<String> getExecutorIdList(String nameSpace, String taskId) throws RuntimeException {
        ZooKeeper zooKeeper = zooKeeperService.getZooKeeper(nameSpace);
        String executorsNodePath = TaskNodePath.getExecutorsNodePath(taskId);
        List<String> executorIds = zooKeeper.getChildren(executorsNodePath);
        if (executorIds == null || CollectionUtils.isEmpty(executorIds)) {
            return Lists.newArrayList();
        }

        return executorIds;
    }

    @Override
    public GetTaskConfigurationVo getTaskConfigurationVo(String nameSpace, String taskId) throws RuntimeException {
        TaskConfigurationDto taskConfiguration4DB = taskConfigurationDtoService.getTaskConfigurationDtoByNameSpaceAndTaskId(nameSpace, taskId);
        if (taskConfiguration4DB == null) {
            throw new RuntimeException(String.format("该作业(%s)不存在", taskId));
        }
        GetTaskConfigurationVo getTaskConfigurationVo = new GetTaskConfigurationVo();
        TaskConfiguration taskConfiguration = new TaskConfiguration();
        BeanUtils.copyProperties(taskConfiguration4DB, taskConfiguration);
        taskConfiguration.setDefaultValues();
        getTaskConfigurationVo.copyFrom(taskConfiguration);

        getTaskConfigurationVo.setTimeZonesProvided(Arrays.asList(TimeZone.getAvailableIDs()));
        getTaskConfigurationVo.setPreferListProvided(getExecutorProvidedList(nameSpace, taskId));


        ZooKeeper zooKeeper = zooKeeperService.getZooKeeper(nameSpace);
        getTaskConfigurationVo.setStatus(getTaskStatus(getTaskConfigurationVo.getJobName(), zooKeeper, getTaskConfigurationVo.getEnabled()));

        return getTaskConfigurationVo;
    }


    @Transactional(rollbackFor = Exception.class)
    @Override
    public void updateTaskConfiguration(String namespace, TaskConfiguration taskConfiguration, String updatedBy)
            throws RuntimeException {
        TaskConfigurationDto oldTaskConfiguration4DB = taskConfigurationDtoService.getTaskConfigurationDtoByNameSpaceAndTaskId(namespace, taskConfiguration.getTaskId());
        if (oldTaskConfiguration4DB == null) {
            throw new RuntimeException(String.format("该作业(%s)不存在", taskConfiguration.getTaskId()));
        }
        // 从数据库拿出老的数据，将需要更新的数据赋值（为空的字段视为不需要更新）
        TaskConfigurationDto newTaskConfiguration4DB = new TaskConfigurationDto();
        BeanUtils.copyProperties(oldTaskConfiguration4DB, newTaskConfiguration4DB);
        BeanUtils.copyPropertiesIgnoreNull(taskConfiguration, newTaskConfiguration4DB);
        // 与老的数据库中的该作业的配置对比，如果没有改变，则直接返回
        if (oldTaskConfiguration4DB.equals(newTaskConfiguration4DB)) {
            return;
        }
        // 设置作业配置字段默认值，并且强制纠正某些字段
        correctConfigValueWhenUpdateJob(newTaskConfiguration4DB);
        // 校验作业配置
        validateTaskConfiguration(newTaskConfiguration4DB);
        // 更新该作业到数据库
        taskConfigurationDtoService.updateNewAndSaveOld2History(newTaskConfiguration4DB, oldTaskConfiguration4DB, updatedBy);

        // 更新作业配置到zk，并联动更新关联作业的上下游
        updateTaskConfigurationToZk(newTaskConfiguration4DB, zooKeeperService.getZooKeeper(namespace));
    }

    private void correctConfigValueWhenUpdateJob(TaskConfiguration taskConfiguration) {
        // 对不符合要求的字段重新设置为默认值
        taskConfiguration.setDefaultValues();
        // 消息作业不failover不rerun
        TaskType taskType = TaskType.getJobType(taskConfiguration.getTaskType());

        // 被动作业不rerun
        if (TaskType.isPassive(taskType)) {
            taskConfiguration.setRerun(false);
        }


    }

    private void updateTaskConfigurationToZk(TaskConfiguration taskConfiguration, ZooKeeper zooKeeper) throws RuntimeException {
        try {
            String taskId = taskConfiguration.getTaskId();

            TransactionalZooKeeper transactionalZooKeeper = zooKeeper.useTransaction();

            // 更新作业
            transactionalZooKeeper
                    .replaceIfChanged(TaskNodePath.getConfigNodePath(taskId, CONFIG_ITEM_ENABLED), taskConfiguration.getEnabled())
                    .replaceIfChanged(TaskNodePath.getConfigNodePath(taskId, CONFIG_ITEM_DESCRIPTION), taskConfiguration.getDescription())
                    .replaceIfChanged(TaskNodePath.getConfigNodePath(taskId, CONFIG_ITEM_JOB_TYPE), taskConfiguration.getTaskType())

                    .replaceIfChanged(TaskNodePath.getConfigNodePath(taskId, CONFIG_ITEM_SHARDING_ITEM_PARAMETERS), taskConfiguration.getTaskItemParameters())
                    .replaceIfChanged(TaskNodePath.getConfigNodePath(taskId, CONFIG_ITEM_JOB_PARAMETER), taskConfiguration.getTaskParameter())

                    .replaceIfChanged(TaskNodePath.getConfigNodePath(taskId, CONFIG_ITEM_FAILOVER), taskConfiguration.getFailover())
                    .replaceIfChanged(TaskNodePath.getConfigNodePath(taskId, CONFIG_ITEM_MONITOR_EXECUTION), "true")
                    .replaceIfChanged(TaskNodePath.getConfigNodePath(taskId, CONFIG_ITEM_TIMEOUT_4_ALARM_SECONDS), taskConfiguration.getTimeout4AlarmSeconds())
                    .replaceIfChanged(TaskNodePath.getConfigNodePath(taskId, CONFIG_ITEM_TIMEOUT_SECONDS), taskConfiguration.getTimeoutSeconds())
                    .replaceIfChanged(TaskNodePath.getConfigNodePath(taskId, CONFIG_ITEM_CRON), taskConfiguration.getCronExpression())
                    .replaceIfChanged(TaskNodePath.getConfigNodePath(taskId, CONFIG_ITEM_PAUSE_PERIOD_DATE), taskConfiguration.getPauseDatePeriod())
                    .replaceIfChanged(TaskNodePath.getConfigNodePath(taskId, CONFIG_ITEM_PAUSE_PERIOD_TIME), taskConfiguration.getPauseTimePeriod())
                    .replaceIfChanged(TaskNodePath.getConfigNodePath(taskId, CONFIG_ITEM_PROCESS_COUNT_INTERVAL_SECONDS), taskConfiguration.getProcessCountIntervalSeconds())
                    .replaceIfChanged(TaskNodePath.getConfigNodePath(taskId, CONFIG_ITEM_SHARDING_TOTAL_COUNT), taskConfiguration.getTaskItemTotalCount())
                    .replaceIfChanged(TaskNodePath.getConfigNodePath(taskId, CONFIG_ITEM_SHOW_NORMAL_LOG), taskConfiguration.getShowNormalLog())
                    .replaceIfChanged(TaskNodePath.getConfigNodePath(taskId, CONFIG_ITEM_LOAD_LEVEL), taskConfiguration.getLoadLevel())
                    .replaceIfChanged(TaskNodePath.getConfigNodePath(taskId, CONFIG_ITEM_PREFER_LIST), taskConfiguration.getPreferList())
                    .replaceIfChanged(TaskNodePath.getConfigNodePath(taskId, CONFIG_ITEM_USE_DISPREFER_LIST), taskConfiguration.getUseDisPreferList())


                    .replaceIfChanged(TaskNodePath.getConfigNodePath(taskId, CONFIG_ITEM_RERUN), taskConfiguration.getRerun())

                    .replaceIfChanged(TaskNodePath.getConfigNodePath(taskId, CONFIG_ITEM_JOB_CLASS), taskConfiguration.getTaskClass());

            // 提交事务
            transactionalZooKeeper.commit();
        } catch (Exception e) {
            log.error("update job to zk failed", e);
            throw new RuntimeException(e);
        }
    }


    @Override
    public List<String> getTaskIdFromZooKeeper(String nameSpace) throws RuntimeException {
        ZooKeeper zooKeeper = zooKeeperService.getZooKeeper(nameSpace);
        String tasksNodePath = TaskNodePath.getTasksNodePath();
        List<String> taskIdList = zooKeeper.getChildren(tasksNodePath);
        if (taskIdList == null) {
            return Lists.newArrayList();
        }

        List<String> $taskIdList = new ArrayList<>();
        for (String taskId : taskIdList) {
            // 如果config节点存在才视为正常作业，其他异常作业在其他功能操作时也忽略
            if (zooKeeper.checkExists(TaskNodePath.getConfigNodePath(taskId))) {
                $taskIdList.add(taskId);
            }
        }
        Collections.sort($taskIdList);
        return $taskIdList;
    }

    @Transactional(rollbackFor = Exception.class)
    @Override
    public void updateTaskCron(String nameSpace, String taskId, String cron, Map<String, String> customContext, String updatedBy) throws RuntimeException {
        String cron0 = cron;
        if (cron0 != null && !cron0.trim().isEmpty()) {
            try {
                cron0 = cron0.trim();
                CronExpression.validateExpression(cron0);
            } catch (ParseException e) {
                throw new RuntimeException("The cron expression is invalid: " + cron);
            }
        } else {
            cron0 = "";
        }
        ZooKeeper zooKeeper = zooKeeperService.getZooKeeper(nameSpace);
        if (zooKeeper.checkExists(TaskNodePath.getConfigNodePath(taskId))) {
            String newCustomContextStr = null;

            String oldCustomContextStr = zooKeeper.getData(TaskNodePath.getConfigNodePath(taskId, CONFIG_ITEM_CUSTOM_CONTEXT));
            Map<String, String> oldCustomContextMap = toCustomContext(oldCustomContextStr);
            if (customContext != null && !customContext.isEmpty()) {
                oldCustomContextMap.putAll(customContext);
                newCustomContextStr = toCustomContext(oldCustomContextMap);
                if (newCustomContextStr.length() > 8000) {
                    throw new RuntimeException("The all customContext is out of db limit (Varchar[8000])");
                }
                if (newCustomContextStr.getBytes().length > 1024 * 1024) {
                    throw new RuntimeException("The all customContext is out of zk limit memory(1M)");
                }
            }

            String newCron = null;
            String oldCron = zooKeeper.getData(TaskNodePath.getConfigNodePath(taskId, CONFIG_ITEM_CRON));
            if (cron0 != null && oldCron != null && !cron0.equals(oldCron.trim())) {
                newCron = cron0;
            }
            if (newCustomContextStr != null || newCron != null) {
                saveCronToDb(taskId, zooKeeper, newCustomContextStr, newCron, updatedBy);
            }
            if (newCustomContextStr != null) {
                zooKeeper.update(TaskNodePath.getConfigNodePath(taskId, CONFIG_ITEM_CUSTOM_CONTEXT), newCustomContextStr);
            }
            if (newCron != null) {
                zooKeeper.update(TaskNodePath.getConfigNodePath(taskId, CONFIG_ITEM_CRON), newCron);
            }
        } else {
            throw new RuntimeException("The job does not exists: " + taskId);
        }
    }

    private void saveCronToDb(String taskId, ZooKeeper zooKeeper, String newCustomContextStr, String newCron, String updatedBy) throws RuntimeException {
        String nameSpace = zooKeeper.getCuratorFramework().getNamespace();
        TaskConfigurationDto oldTaskConfiguration4DB = taskConfigurationDtoService.getTaskConfigurationDtoByNameSpaceAndTaskId(nameSpace, taskId);
        if (oldTaskConfiguration4DB == null) {
            String errorMsg = "在DB找不到该作业的配置, namespace：" + nameSpace + " jobName:" + taskId;
            log.error(errorMsg);
            throw new RuntimeException(errorMsg);
        }
        TaskConfigurationDto newTaskConfiguration4DB = new TaskConfigurationDto();
        BeanUtils.copyProperties(oldTaskConfiguration4DB, newTaskConfiguration4DB);

        if (newCron != null) {
            newTaskConfiguration4DB.setCronExpression(newCron);
        }
        taskConfigurationDtoService.updateNewAndSaveOld2History(newTaskConfiguration4DB, oldTaskConfiguration4DB, updatedBy);
    }

    /**
     * 将str转为map
     *
     * @param customContextStr str字符串
     * @return 自定义上下文map
     */
    private Map<String, String> toCustomContext(String customContextStr) {
        Map<String, String> customContext = null;
        if (customContextStr != null) {
            //customContext = JsonUtils.fromJSON(customContextStr, customContextType);
        }
        if (customContext == null) {
            customContext = new HashMap<>();
        }
        return customContext;
    }

    /**
     * 将map转为str字符串
     *
     * @param customContextMap 自定义上下文map
     * @return 自定义上下文str
     */
    private String toCustomContext(Map<String, String> customContextMap) {
        String result = null;//JsonUtils.toJSON(customContextMap);
        if (result == null) {
            result = "";
        }
        return result.trim();
    }

    @Override
    public List<TaskExecutor> getTaskExecutorList(String nameSpace, String taskId) throws RuntimeException {
        ZooKeeper zooKeeper = zooKeeperService.getZooKeeper(nameSpace);
        String executorsNodePath = TaskNodePath.getExecutorsNodePath(taskId);
        List<String> executorIdList = zooKeeper.getChildren(executorsNodePath);
        List<TaskExecutor> taskExecutorList = new ArrayList<>();
        if (executorIdList != null && !executorIdList.isEmpty()) {
            String leaderIp = zooKeeper.getData(TaskNodePath.getLeaderNodePath(taskId, "election/host"));
            TaskStatus taskStatus = getTaskStatus(nameSpace, taskId);
            for (String executorId : executorIdList) {
                TaskExecutor taskExecutor = getTaskExecutor(taskId, leaderIp, executorId, zooKeeper);
                taskExecutor.setTaskStatus(taskStatus);
                taskExecutorList.add(taskExecutor);
            }
        }
        return taskExecutorList;
    }

    @Override
    public List<TaskExecutorStatus> getTaskExecutorStatusList(String nameSpace, String taskId) throws RuntimeException {
        ZooKeeper zooKeeper = zooKeeperService.getZooKeeper(nameSpace);
        List<String> executorIdList = getExecutorIdList(nameSpace, taskId);
        List<TaskExecutorStatus> taskExecutorStatusList = new ArrayList<>();
        if (executorIdList != null && !executorIdList.isEmpty()) {
            for (String executorId : executorIdList) {
                taskExecutorStatusList.add(getTaskExecutorStatus(taskId, executorId, zooKeeper));
            }
        }

        return taskExecutorStatusList;
    }

    private TaskExecutorStatus getTaskExecutorStatus(String taskId, String executorId, ZooKeeper zooKeeper) {
        TaskExecutorStatus taskExecutorStatus = new TaskExecutorStatus();
        taskExecutorStatus.setExecutorName(executorId);
        taskExecutorStatus.setTaskId(taskId);
        taskExecutorStatus.setTaskExecutorStatus12(getTaskExecutorStatus12(taskId, executorId, zooKeeper));
        return taskExecutorStatus;
    }

    private TaskExecutorStatus12 getTaskExecutorStatus12(String taskId, String executorId, ZooKeeper zooKeeper) {
        String status = zooKeeper.getData(TaskNodePath.getExecutorsNodePath(taskId, executorId, "status"));
        return TaskExecutorStatus12.getExecutorStatus(status);
    }

    private TaskExecutor getTaskExecutor(String taskId, String leaderIp, String executorId, ZooKeeper zooKeeper) {
        TaskExecutor taskExecutor = new TaskExecutor();
        taskExecutor.setExecutorId(executorId);
        taskExecutor.setIp(zooKeeper.getData(TaskNodePath.getExecutorsNodePath(taskId, executorId, "ip")));
        taskExecutor.setVersion(zooKeeper.getData(TaskNodePath.getExecutorsNodePath(taskId, executorId, "version")));
        String processSuccessCount = zooKeeper.getData(TaskNodePath.getExecutorsNodePath(taskId, executorId, "processSuccessCount"));
        taskExecutor.setProcessSuccessCount(null == processSuccessCount ? 0 : Integer.parseInt(processSuccessCount));
        String processFailureCount = zooKeeper.getData(TaskNodePath.getExecutorsNodePath(taskId, executorId, "processFailureCount"));
        taskExecutor.setProcessFailureCount(null == processFailureCount ? 0 : Integer.parseInt(processFailureCount));
        taskExecutor.setSharding(zooKeeper.getData(TaskNodePath.getExecutorsNodePath(taskId, executorId, "sharding")));
        taskExecutor.setTaskExecutorStatus12(getTaskExecutorStatus12(taskId, executorId, zooKeeper));
        taskExecutor.setLeader(executorId.equals(leaderIp));
        taskExecutor.setJobVersion(getJobVersion(taskId, executorId, zooKeeper));
        taskExecutor.setContainer(zooKeeper.checkExists(ExecutorNodePath.getExecutorTaskNodePath(executorId)));

        return taskExecutor;
    }

    private String getJobVersion(String taskId, String executorId, ZooKeeper zooKeeper) {
        String jobVersion = zooKeeper.getData(TaskNodePath.getExecutorsNodePath(taskId, executorId, "jobVersion"));
        return jobVersion == null ? "" : jobVersion;
    }

    @Override
    public void runAtOnce(String nameSpace, String taskId) throws RuntimeException {
        TaskStatus taskStatus = getTaskStatus(nameSpace, taskId);
        if (!TaskStatus.READY.equals(taskStatus)) {
            throw new RuntimeException(String.format("该作业(%s)不处于READY状态，不能立即执行", taskId));
        }
        List<TaskExecutorStatus> jobServerStatusList = getTaskExecutorStatusList(nameSpace, taskId);
        if (jobServerStatusList != null && !jobServerStatusList.isEmpty()) {
            boolean hasOnlineExecutor = false;
            ZooKeeper zooKeeper = zooKeeperService.getZooKeeper(nameSpace);
            for (TaskExecutorStatus taskExecutorStatus : jobServerStatusList) {
                if (TaskExecutorStatus12.ONLINE.equals(taskExecutorStatus.getTaskExecutorStatus12())) {
                    hasOnlineExecutor = true;
                    String executorName = taskExecutorStatus.getExecutorName();
                    String path = TaskNodePath.getRunOneTimePath(taskId, executorName);
                    if (zooKeeper.checkExists(path)) {
                        zooKeeper.delete(path);
                    }
                    zooKeeper.create(path, "null");
                    log.info("runAtOnce namespace:{}, jobName:{}, executorName:{}", nameSpace, taskId, executorName);
                }
            }
            if (!hasOnlineExecutor) {
                throw new RuntimeException("没有ONLINE的executor，不能立即执行");
            }
        } else {
            throw new RuntimeException(String.format("没有executor接管该作业(%s)，不能立即执行", taskId));
        }
    }

    @Override
    public void stopAtOnce(String nameSpace, String taskId) throws RuntimeException {
        TaskStatus taskStatus = getTaskStatus(nameSpace, taskId);
        if (!TaskStatus.STOPPING.equals(taskStatus)) {
            throw new RuntimeException(String.format("该作业(%s)不处于STOPPING状态，不能立即终止", taskId));
        }
        List<String> executorIdList = getExecutorIdList(nameSpace, taskId);
        if (executorIdList != null && !executorIdList.isEmpty()) {
            ZooKeeper zooKeeper = zooKeeperService.getZooKeeper(nameSpace);
            for (String executorName : executorIdList) {
                String path = TaskNodePath.getStopOneTimePath(taskId, executorName);
                if (zooKeeper.checkExists(path)) {
                    zooKeeper.delete(path);
                }
                zooKeeper.create(path);
                log.info("stopAtOnce namespace:{}, jobName:{}, executorName:{}", nameSpace, taskId, executorName);
            }
        } else {
            throw new RuntimeException(String.format("没有executor接管该作业(%s)，不能立即终止", taskId));
        }
    }

    @Override
    public List<ExecutionInfo> getExecutionStatus(String nameSpace, String taskId) throws RuntimeException {
        ZooKeeper zooKeeper = zooKeeperService.getZooKeeper(nameSpace);
        TaskConfiguration taskConfiguration = getTaskConfiguration(nameSpace, taskId);
        if (!taskConfiguration.getEnabled() && TaskStatus.STOPPED.equals(getTaskStatus(taskId, zooKeeper, false))) {
            return Lists.newArrayList();
        }

        // update report node and sleep for 500ms
        updateReportNodeAndWait(taskId, zooKeeper, 500L);
        // 如果execution节点不存在则返回空List
        String executionNodePath = TaskNodePath.getExecutionNodePath(taskId);
        List<String> taskItemList = zooKeeper.getChildren(executionNodePath);
        if (taskItemList == null || taskItemList.isEmpty()) {
            return Lists.newArrayList();
        }

        List<ExecutionInfo> executionInfoList = Lists.newArrayList();
        Map<String, String> itemExecutorMap = buildItem2ExecutorMap(taskId, zooKeeper);
        for (Map.Entry<String, String> itemExecutorEntry : itemExecutorMap.entrySet()) {
            executionInfoList.add(buildExecutionInfo(taskId, itemExecutorEntry.getKey(), itemExecutorEntry.getValue(), zooKeeper, taskConfiguration));
        }

        // 可能有漏掉的running分片，比如新的机器接管了failover分片
        for (String taskItem : taskItemList) {
            if (itemExecutorMap.containsKey(taskItem)) {
                // 已经在之前的步骤计算了
                continue;
            }
            String runningNodePath = TaskNodePath.getExecutionNodePath(taskId, taskItem, "running");
            boolean running = zooKeeper.checkExists(runningNodePath);
            if (running) {
                executionInfoList.add(buildExecutionInfo(taskId, taskItem, null, zooKeeper, taskConfiguration));
            }
        }

        Collections.sort(executionInfoList);

        return executionInfoList;
    }

    @Override
    public String getExecutionLog(String nameSpace, String jobName, String jobItem) throws RuntimeException {
        ZooKeeper zooKeeper = zooKeeperService.getZooKeeper(nameSpace);
        String jobLogNodePath = TaskNodePath.getExecutionNodePath(jobName, jobItem, "jobLog");
        Stat stat = zooKeeper.getStat(jobLogNodePath);
        if (stat.getDataLength() > getMaxZooKeeperNodeDataLength()) {
            log.warn("job log of job={} item={} exceed max length, will not display the original log", jobName, jobItem);
            return ERR_MSG_TOO_LONG_TO_DISPLAY;
        }

        return zooKeeper.getData(jobLogNodePath);
    }


    private void updateReportNodeAndWait(String nameSpace, ZooKeeper zooKeeper, long sleepInMill) {
        zooKeeper.update(TaskNodePath.getReportPath(nameSpace), System.currentTimeMillis());
        try {
            Thread.sleep(sleepInMill);
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }

    private ExecutionInfo buildExecutionInfo(String taskId, String taskItem, String executorName, ZooKeeper zooKeeper, TaskConfiguration taskConfiguration) {
        ExecutionInfo executionInfo = new ExecutionInfo();
        executionInfo.setTaskId(taskId);
        executionInfo.setTaskItem(Integer.parseInt(taskItem));

        setExecutorNameAndStatus(taskId, taskItem, executorName, zooKeeper, executionInfo, taskConfiguration);

        // jobMsg
        String jobMsg = zooKeeper.getData(TaskNodePath.getExecutionNodePath(taskId, taskItem, "jobMsg"));
        executionInfo.setJobMsg(jobMsg);

        // timeZone


        // last begin time
        String lastBeginTime = zooKeeper.getData(TaskNodePath.getExecutionNodePath(taskId, taskItem, "lastBeginTime"));
        executionInfo.setLastBeginTime(lastBeginTime);
        // next fire time, ignore if jobType is Msg
        TaskType jobType = TaskType.getJobType(taskConfiguration.getTaskType());
        if (TaskType.isCron(jobType)) {
            String nextFireTime = zooKeeper.getData(TaskNodePath.getExecutionNodePath(taskId, taskItem, "nextFireTime"));
            executionInfo.setNextFireTime(nextFireTime);
        } else {
            executionInfo.setNextFireTime(null);
        }
        // last complete time
        String lastCompleteTime = zooKeeper.getData(TaskNodePath.getExecutionNodePath(taskId, taskItem, "lastCompleteTime"));
        if (lastCompleteTime != null) {
            long lastCompleteTimeLong = Long.parseLong(lastCompleteTime);
            if (lastBeginTime == null) {
                executionInfo.setLastCompleteTime(lastCompleteTime);
            } else {
                long lastBeginTimeLong = Long.parseLong(lastBeginTime);
                if (lastCompleteTimeLong >= lastBeginTimeLong) {
                    executionInfo.setLastCompleteTime(lastCompleteTime);

                    executionInfo.setLastTimeConsumedInSec((lastCompleteTimeLong - lastBeginTimeLong) / 1000d);
                }
            }
        }

        return executionInfo;
    }

    private void setExecutorNameAndStatus(String taskId, String taskItem, String executorId, ZooKeeper zooKeeper, ExecutionInfo executionInfo, TaskConfiguration taskConfiguration) {


        boolean isCompleted = false;
        String completedNodePath = TaskNodePath.getCompletedNodePath(taskId, taskItem);
        String completedData = zooKeeper.getData(completedNodePath);
        if (completedData != null) {
            isCompleted = true;
            executionInfo.setExecutorId(StringUtils.isNotBlank(completedData) ? completedData : executorId);
            // 不能立即返回还是要看看是否failed或者timeout
        }

        String failedNodePath = TaskNodePath.getFailedNodePath(taskId, taskItem);
        if (zooKeeper.checkExists(failedNodePath)) {
            if (isCompleted) {
                executionInfo.setExecutionStatus(ExecutionStatus.FAILED);
            } else {
                log.warn(ERR_MSG_PENDING_STATUS, taskId, taskItem, executorId, "no completed node found but only failed node");
                executionInfo.setExecutorId(executorId);
                executionInfo.setExecutionStatus(ExecutionStatus.PENDING);
            }
            return;
        }

        String timeoutNodePath = TaskNodePath.getTimeoutNodePath(taskId, taskItem);
        if (zooKeeper.checkExists(timeoutNodePath)) {
            if (isCompleted) {
                executionInfo.setExecutionStatus(ExecutionStatus.TIMEOUT);
            } else {
                log.warn(ERR_MSG_PENDING_STATUS, taskId, taskItem, executorId, "no completed node found but only timeout node");
                executionInfo.setExecutorId(executorId);
                executionInfo.setExecutionStatus(ExecutionStatus.PENDING);
            }
            return;
        }

        // 只有completed节点没有timeout/failed意味着成功，于是立即返回
        if (isCompleted) {
            executionInfo.setExecutionStatus(ExecutionStatus.COMPLETED);
            return;
        }

        boolean isRunning = false;
        String runningNodePath = TaskNodePath.getRunningNodePath(taskId, taskItem);
        String runningData = zooKeeper.getData(runningNodePath);
        if (runningData != null) {
            isRunning = true;
            executionInfo.setExecutorId(StringUtils.isBlank(runningData) ? executorId : runningData);
            long mtime = zooKeeper.getMtime(runningNodePath);
            executionInfo.setTimeConsumed((new Date().getTime() - mtime) / 1000);
            executionInfo.setExecutionStatus(ExecutionStatus.RUNNING);
            // 不能立即返回还是要看看是否正在failover
        }

        String failoverNodePath = TaskNodePath.getFailoverNodePath(taskId, taskItem);
        String failoverData = zooKeeper.getData(failoverNodePath);
        if (failoverData != null) {
            // 设置为failover节点的真是executorName
            executionInfo.setExecutorId(failoverData);
            executionInfo.setFailover(true);
            // 如果有failover节点，running应该配对出现，否则显示pending状态
            if (!isRunning) {
                log.warn(ERR_MSG_PENDING_STATUS, taskId, taskItem, executorId, "no running node found but only failover node");
                executionInfo.setExecutionStatus(ExecutionStatus.PENDING);
            }

            return;
        }

        if (!isRunning) {
            log.warn(ERR_MSG_PENDING_STATUS, taskId, taskItem, executorId, "no running node or completed node found");
            executionInfo.setExecutionStatus(ExecutionStatus.PENDING);
        }
    }

    private Map<String, String> buildItem2ExecutorMap(String taskId, ZooKeeper zooKeeper) {

        String executorsNodePath = TaskNodePath.getExecutorsNodePath(taskId);
        List<String> executorList = zooKeeper.getChildren(executorsNodePath);

        if (executorList == null || executorList.isEmpty()) {
            return Maps.newHashMap();
        }

        Map<String, String> resultMap = new HashMap<>();
        for (String server : executorList) {
            resolveShardingData(taskId, zooKeeper, resultMap, server);
        }
        return resultMap;
    }

    private void resolveShardingData(String taskId, ZooKeeper zooKeeper, Map<String, String> resultMap, String server) {
        String shardingData = zooKeeper.getData(TaskNodePath.getServerSharding(taskId, server));
        if (StringUtils.isBlank(shardingData)) {
            return;
        }

        String[] shardingValues = shardingData.split(",");
        for (String value : shardingValues) {
            if (StringUtils.isBlank(value)) {
                continue;
            }

            resultMap.put(value.trim(), server);
        }
    }


}
